Skip to content

Rabbitmq eventbus buffer implementation - #14677

Open
durdin85 wants to merge 3 commits into
apache:mainfrom
durdin85:rabbitmq-eventbus-buffer
Open

durdin85 wants to merge 3 commits into
apache:mainfrom
durdin85:rabbitmq-eventbus-buffer

Conversation

@durdin85

Copy link
Copy Markdown

Description

This PR makes the RabbitMQ event bus keep events that are published while the
AMQP broker is unreachable and send them, in the order they were published, once
the connection is back. Until now such events were discarded.

Behaviour:

  • Events that cannot be sent are kept in an in-memory FIFO buffer (routing key,
    body, time of buffering).
  • Buffering starts only after the broker was reachable at least once. Before
    that, publish() fails as it does today.
  • The buffer is bounded: at most bufferSize events (default 1000, the oldest is
    dropped when full) and each event is dropped bufferExpiry milliseconds
    (default 300000) after it was buffered. bufferSize 0 disables buffering and
    restores the previous behaviour.
  • While the buffer is not empty, or the connection is down, new events are
    appended to the buffer, so they never overtake older ones. publish() returns
    normally for a buffered event and no longer connects on the calling thread
    while the broker is down (the connection timeout of amqp-client 5.17.0 is
    60 s).
  • ReconnectionTask now runs for the lifetime of the event bus instead of being
    started by DisconnectHandler. Every retryInterval it drops expired events,
    reconnects if there is no connection and sends the buffered events. This also
    re-establishes connections closed by the bus itself (AlreadyClosedException
    in publish(), blocked connection), which previously started no reconnect.
  • stop() shuts the executor down so the task ends with the bus.

Both limits are optional properties of the existing bean, next to the other AMQP
settings:

<bean id="eventNotificationBus" class="org.apache.cloudstack.mom.rabbitmq.RabbitMQEventBus">
    ...
    <property name="bufferSize" value="1000"/>
    <property name="bufferExpiry" value="300000"/>
</bean>

Known limits, by design:

  • No publisher confirms are used. An event written to the socket just before the
    connection breaks can still be lost unnoticed.
  • An event whose send fails after the broker already received it is sent again
    after the reconnect (at-least-once for buffered events).
  • The buffer is in memory only and is lost when the management server restarts.

No new dependencies; the change is confined to RabbitMQEventBus.

Fixes: #14676

AI disclosure: The code, the unit test and this description were generated with an AI assistant. The design and its constraints were set by the submitter and the generated code had been reviewed by the submitter. If the project does not accept AI-generated contributions, please say so and treat this as a proposal for discussion.

Types of changes

  • Breaking change (fix or feature that would cause existing functionality to change)
  • New feature (non-breaking change which adds functionality)
  • Bug fix (non-breaking change which fixes an issue)
  • Enhancement (improves an existing feature and functionality)
  • Cleanup (Code refactoring and cleanup, that may add test cases)
  • Build/CI
  • Test (unit or integration test code)

Feature/Enhancement Scale or Bug Severity

Feature/Enhancement Scale

  • Major
  • Minor

How Has This Been Tested?

New unit test RabbitMQEventBusTest (mocked AMQP connection and channel):

  • publish() still fails when the broker was never reachable;
  • events published while disconnected are sent in order after the reconnect, and
    an event published while the buffer is not empty does not overtake them;
  • an event whose send fails with an IOException is buffered and re-sent;
  • the oldest event is dropped when the buffer is full;
  • expired events are dropped, also while disconnected.

The plugin builds and RabbitMQEventBusTest passes on main and on the 4.22 branch (mvn -pl plugins/event-bus/rabbitmq -am test -Dtest=RabbitMQEventBusTest).

Not yet tested against a real RabbitMQ broker.

How did you try to break this feature and the system with this change?

The unit tests cover a full buffer, expired events, a failing send and a
connection that returns while events are still buffered.

durdin85 and others added 2 commits November 11, 2026 11:00
Events published while the AMQP connection is down were discarded.
Keep them in a bounded in-memory FIFO buffer and send them in order
once the connection is back.

- events are buffered only after the broker was reachable once
- buffer is capped (bufferSize, default 1000, oldest dropped first)
  and entries expire (bufferExpiry, default 300000 ms)
- publish() no longer connects on the calling thread while the
  broker is down
- ReconnectionTask runs for the lifetime of the bus, so connections
  closed by the bus itself are re-established as well

This change, including the unit test, was generated with an AI
assistant (Claude Code).

Fixes: apache#14676

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
- keep the buffer flush inside the try block, so an unexpected
  exception cannot end the only reconnect thread
- check for an existing connection and connect under the same lock
  as getConnection(), so a connection created by publish() is not
  replaced and leaked

Generated with an AI assistant (Claude Code).

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
@boring-cyborg

boring-cyborg Bot commented Nov 11, 2026

Copy link
Copy Markdown

Congratulations on your first Pull Request and welcome to the Apache CloudStack community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/cloudstack/blob/main/CONTRIBUTING.md)
Here are some useful points:

Generated with an AI assistant (Claude Code).

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Feature: AMQP event bus - buffer events while the broker is unreachable

1 participant